
package com.sdk.apiNotify.rabbitmq.consumber;

import com.coment.dto.ApiNotify;
import com.rabbitmq.client.Channel;
import com.sdk.apiNotify.rabbitmq.consumber.server.ApiNotifyPushService;
import com.sdk.apiNotify.rabbitmq.producer.ApiNotifyProducer;
import com.sdk.service.impl.ApiNotifyServiceImpl;
import lombok.RequiredArgsConstructor;
import lombok.extern.slf4j.Slf4j;
import org.springframework.amqp.core.Message;
import org.springframework.amqp.rabbit.annotation.RabbitListener;
import org.springframework.amqp.support.converter.MessageConverter;
import org.springframework.stereotype.Service;

@Slf4j
@Service
@RequiredArgsConstructor
public class ApiNotifyConsumer2 {
    private final ApiNotifyProducer apiNotifyProducer;
    private final ApiNotifyPushService apiNotifyPushService;
    private final ApiNotifyServiceImpl apiNotifyServiceImpl;
    // 使用注入的MessageConverter而不是自己创建实例
    private final MessageConverter messageConverter;

    /**
     * 抖音API通知监听器
     */
    @RabbitListener(
            queues = "#{topicCfg.topic_dy}",
            ackMode = "MANUAL",
            concurrency = "5"
    )
    public void handleDouyinApiNotify(Message message, Channel channel) throws Exception {
        handleMessage(message, channel);
    }

    /**
     * 快手API通知监听器
     */
    @RabbitListener(
            queues = "#{topicCfg.topic_ks}",
            ackMode = "MANUAL",
            concurrency = "5"
    )
    public void handleKuaiShouApiNotify(Message message, Channel channel) throws Exception {
        handleMessage(message, channel);
    }

    /**
     * 重试队列监听器
     */
    @RabbitListener(
            queues = "#{topicCfg.topicRetry}",
            ackMode = "MANUAL",
            concurrency = "3"
    )
    public void handleRetryMessage(Message message, Channel channel) throws Exception {
        ApiNotify apiNotify;
        try {
            apiNotify = (ApiNotify) messageConverter.fromMessage(message);
        } catch (Exception e) {
            log.error("重试消息解析失败", e);
            // 消息格式错误，直接确认避免重复消费
            channel.basicAck(message.getMessageProperties().getDeliveryTag(), false);
            return;
        }

        Integer retryCount = apiNotify.getRetryCount();
        retryCount = retryCount + 1;
        apiNotify.setRetryCount(retryCount);

        try {
            log.warn("第 {} 次重试 {}", retryCount, apiNotify);
            this.apiNotifyPushService.handlerMessage(apiNotify);
            // 手动确认消息
            channel.basicAck(message.getMessageProperties().getDeliveryTag(), false);
        } catch (Exception e) {
            log.error("重试处理失败，消息ID: {}", apiNotify.getMessageId(), e);
            if (retryCount < 3) {
                this.apiNotifyProducer.sendToRetryQueueWithHeaders(apiNotify);
            } else {
                this.apiNotifyProducer.sendToDeadLetterQueue(apiNotify);
            }
            // 手动确认重试消息，防止重复消费
            channel.basicAck(message.getMessageProperties().getDeliveryTag(), false);
        }
    }

    /**
     * 死信队列监听器
     */
    @RabbitListener(
            queues = "#{topicCfg.topicDlx}",
            ackMode = "MANUAL", // 改为手动确认以确保处理完成
            concurrency = "1"
    )
    public void handleDlxMessage(Message message, Channel channel) throws Exception {
        try {
            ApiNotify apiNotify = (ApiNotify) messageConverter.fromMessage(message);
            log.error("死信队列消息{}，最终处理失败,次数: {}", apiNotify.getMessageId(), apiNotify.getRetryCount());
            apiNotify.setStatus(2);
            this.apiNotifyServiceImpl.saveApiNotify(apiNotify);
            // 手动确认死信消息处理完成
            channel.basicAck(message.getMessageProperties().getDeliveryTag(), false);
        } catch (Exception e) {
            log.error("处理死信队列消息失败", e);
            // 死信消息处理失败也应确认，避免阻塞死信队列
            channel.basicAck(message.getMessageProperties().getDeliveryTag(), false);
        }
    }

    /**
     * 统一处理抖音和快手的API通知消息
     */
    private void handleMessage(Message message, Channel channel) throws Exception {
        ApiNotify apiNotify;
        try {
            apiNotify = (ApiNotify) messageConverter.fromMessage(message);
        } catch (Exception e) {
            log.error("处理消息失败", e);
            // 消息格式错误，拒绝并不重新入队
            channel.basicNack(message.getMessageProperties().getDeliveryTag(), false, false);
            return;
        }

        long deliveryTag = message.getMessageProperties().getDeliveryTag();

        try {
            this.apiNotifyPushService.handlerMessage(apiNotify);
            apiNotify.setStatus(1);
            this.apiNotifyServiceImpl.saveApiNotify(apiNotify);
            channel.basicAck(deliveryTag, false);
        } catch (Exception e) {
            log.error("消息处理失败，消息ID: {}", apiNotify.getMessageId(), e);
            channel.basicNack(deliveryTag, false, false);
            apiNotify.setRetryCount(0);
            apiNotify.setStatus(0);
            this.apiNotifyProducer.sendToRetryQueue(apiNotify);
            this.apiNotifyServiceImpl.saveApiNotify(apiNotify);
        }
    }
}